﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.ComponentModel.Composition;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Thrift;
using Thrift.Protocol;
using Thrift.Server;
using Thrift.Transport;

namespace Dubbo.Rpc.Thrift
{
    [Export(typeof(IProtocol))]
    [ExportMetadata("Name", "thrift")]
    public class ThriftProtocol : AbstractProtocol
    {
        public const int DEFAULT_PORT = 9090;
        public const string IFACE = "Iface";
        public const string PROCESSOR = "Processor";
        public const string CLIENT = "Client";

        //private static readonly Logger logger = LoggerFactory.getLogger(Thrift2Protocol.class);

        private readonly ConcurrentDictionary<String, TServer> serverDict = new ConcurrentDictionary<String, TServer>();

        [Import]
        public IProxyFactory ProxyFactory { get; set; }

        public override int GetDefaultPort()
        {
            return DEFAULT_PORT;
        }
        
        public override IExporter<T> Export<T>(IInvoker<T> invoker)
        {
            string key = invoker.Url.Address;
            if (!serverDict.ContainsKey(key))
            {
                serverDict.AddOrUpdate(key, GetServer(ProxyFactory.GetProxy(invoker), invoker.Url), (k, v) => v);
            }
            // export service.
            key = serviceKey(invoker.Url);
            DefaultExporter<T> exporter = new DefaultExporter<T>(invoker, key, exporterDict);
            exporterDict.AddOrUpdate(key, exporter, (k, v) => exporter);
            return exporter;
        }

        class DefaultExporter<T> : AbstractExporter<T>
        {
            private readonly string key;
            private readonly ConcurrentDictionary<string, IExporter> exporterDict;

            public DefaultExporter(IInvoker<T> invoker, string key,
                ConcurrentDictionary<string, IExporter> exporterDict)
                : base(invoker)
            {
                this.key = key;
                this.exporterDict = exporterDict;
            }

            public override void Unexport()
            {
                base.Unexport();
                IExporter value;
                exporterDict.TryRemove(key, out value);
            }

        }

        private TServer GetServer<T>(T impl, URL url)
        {
            string typeName = typeof(T).FullName;
            TProcessor tprocessor;
            TServerTransport transport;
            if (typeName.EndsWith(IFACE))
            {
                string processorClsName = typeName.Substring(0, typeName.IndexOf(IFACE)) + PROCESSOR;
                try
                {
                    Type type = Type.GetType(processorClsName + "," + typeof(T).Assembly.FullName);
                    ConstructorInfo ctr = type.GetConstructor(new Type[] { typeof(T) });
                    tprocessor = (TProcessor)ctr.Invoke(new object[] { impl });
                    transport = new TServerSocket(url.Port);

                    TServer server = new TThreadPoolServer(tprocessor, transport);
                    Task.Run(() => server.Serve());
                    return server;
                }
                catch (Exception e)
                {
                    //logger.error(e.getMessage(), e);
                    throw new RpcException("Fail to create thrift server(" + url + ") : " + e.Message, e);
                }
            }
            else
            {
                throw new RpcException("Fail to create thrift server(" + url + ") : " + typeName);
            }
        }

        public override IInvoker<T> Refer<T>(URL url)
        {
            IInvoker<T> invoker = new ThriftInvoker<T>(ProxyFactory, url, GetSockets(url), invokers);
            invokers.Add(invoker);
            return invoker;
        }

        private TSocket[] GetSockets(URL url)
        {
            //int connections = url.GetParameter(Constants.CONNECTIONS_KEY, 1);
            int connections = 3;

            TSocket[] sockets = new TSocket[connections];

            for (int i = 0; i < sockets.Length; i++)
            {
                var socket = new TSocket(url.Host, url.Port);
                //socket.Open();
                sockets[i] = socket;
            }
            return sockets;
        }
    }
}
